package com.lakudouzi.flink.cdc;



import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.springframework.stereotype.Component;
import cn.hutool.extra.spring.SpringUtil;

import java.io.IOException;
import java.util.HashMap;

@Slf4j
@Component
public class DataChangeSink implements SinkFunction<DataChangeInfo> {



    //捕获数据变化，具体执行操作

    @Override
    public void invoke(DataChangeInfo dataChangeInfo, Context context) throws IOException {
        // 变更类型： 0 初始化 1新增 2修改 3删除 4导致源中的现有表被截断的操作

        String data = dataChangeInfo.getData();
        String tableName = dataChangeInfo.getTableName();
        String finalTableName = tableName;
        ElasticsearchClient client = SpringUtil.getBean(ElasticsearchClient.class);
        HashMap dataMap = JSON.parseObject(data, HashMap.class);

            Integer operatorType = dataChangeInfo.getOperatorType();
            switch (operatorType) {
                case 0:


                    log.info("初始化数据:" + data);
                    // 全量数据 更新时启动
//                    client.index(i -> i
//                            .index(finalTableName)
//                            .id(String.valueOf(dataMap.get("id")))
//                            .document(dataMap));
                    break;
                case 1:
                    log.info("新增数据:" + data);
                    client.index(i -> i
                            .index(finalTableName)
                            .id(String.valueOf(dataMap.get("id")))
                            .document(dataMap));
                    break;
                case 2:
                    log.info("修改数据:" + data);
                    client.index(i -> i
                            .index(finalTableName)
                            .id(String.valueOf(dataMap.get("id")))
                            .document(dataMap));
                    break;
                case 3:
                    log.info("删除数据:" + data);
                    client.delete(d -> d.index(finalTableName)
                            .id(String.valueOf(dataMap.get("id"))));
                    break;
                default:
                    log.info("其他数据:" + data);
            }



        // TODO 数据处理,不能在方法外注入需要的bean，会报错必须实例化才可以
        //  (Caused by: org.apache.flink.api.common.InvalidProgramException:  xxx... is not serializable.
        //  The object probably contains or references non serializable fields.)，
        // 所以使用SpringUtil 获取需要的 bean，比如获取 extends ElasticsearchRepository<T, ID>的接口如下所示，
        // 然后就可以使用封装的方法进行增删改操作了

        // 初始化/新增/修改（非逻辑删除）
//        bean.save(systemDept);
        // 删除/修改（逻辑删除）
        // bean.delete(systemDept);
    }




}
